package com.camemax.controller;

import com.camemax.pojo.ItemViewCount;
import com.camemax.pojo.UserBehavior;
import com.camemax.utils.HotItemOperators;
import com.camemax.utils.StreamEnvUtils;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import java.util.Properties;

@RestController
public class HotItemsController {

    /*
    * 每五分钟获取过去一小时的热销商品 => 滑动时间窗口，滑动距离5min，时间窗口60min
    * */
    @GetMapping("/getHotItemsSort/{topNumber}")
    public void getHotItems(@PathVariable("topNumber")int topNumber) throws Exception {
        StreamExecutionEnvironment streamEnv = StreamEnvUtils.getStreamEnv(1);
        DataStream<String> fileDataStream = streamEnv.readTextFile("E:\\Javas\\BigData-Projects\\Data\\UserBehavior.csv");

        // 添加时间戳分配器
        DataStream<UserBehavior> watermarkDataStream = fileDataStream.map(line -> {
            String[] fields = line.split(",");
            return new UserBehavior(Long.valueOf(fields[0]), Long.valueOf(fields[1]), Integer.valueOf(fields[2]), fields[3], Long.valueOf(fields[4]));
        }).assignTimestampsAndWatermarks(
                WatermarkStrategy.<UserBehavior>noWatermarks()
                        .withTimestampAssigner((SerializableTimestampAssigner<UserBehavior>) (element, recordTimestamp) -> element.getTimestamp() * 1000L)
        );

        // 分组聚合开窗
        DataStream<ItemViewCount> windowAggregateStream = watermarkDataStream.filter(data -> "pv".equals(data.getBehavior()))
                .keyBy(UserBehavior::getItemId)
                .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5))) // 滑动窗口实现以每小时分桶、每五分钟滑动
                // aggregate方法Prefix部分 —— 聚合函数
                // aggregate方法Suffix部分 —— 窗口函数
                .aggregate(new HotItemOperators.UserBehaviorAggregateFunctionImpl(),
                        new HotItemOperators.ItemViewCountWindowFunctionImpl());


        SingleOutputStreamOperator<String> resultDataStream = windowAggregateStream.keyBy(ItemViewCount::getTimstamp)
                .process(new HotItemOperators.ItemViewCountKeyedProcessFunctionImpl(topNumber));

        resultDataStream.print();

        streamEnv.execute();
    }

//    @GetMapping("/getKafkaHotItemsSort/{topNumber}")
//    public void getKafkaHotItemsSort(@PathVariable("topNumber")int topNumber) throws Exception {
//
//        StreamExecutionEnvironment streamEnv = StreamEnvUtils.getStreamEnv(1);
//
//        Properties props = new Properties();
//        props.setProperty("bootstrap.servers", "172.28.40.190:9092");
//        props.setProperty("group.id","consumer");
//        props.setProperty("auto.offset.reset","latest");
//
//        DataStreamSource<String> kafkaDataStream = streamEnv.addSource(new FlinkKafkaConsumer<>(
//                "hotItems",
//                new SimpleStringSchema(),
//                props
//        ));
//
//        kafkaDataStream.print();
//
//        // 添加时间戳分配器
//        DataStream<UserBehavior> watermarkDataStream = kafkaDataStream.map(line -> {
//            String[] fields = line.split(",");
//            return new UserBehavior(Long.valueOf(fields[0]), Long.valueOf(fields[1]), Integer.valueOf(fields[2]), fields[3], Long.valueOf(fields[4]));
//        }).assignTimestampsAndWatermarks(
//                WatermarkStrategy.<UserBehavior>noWatermarks()
//                        .withTimestampAssigner((SerializableTimestampAssigner<UserBehavior>) (element, recordTimestamp) -> element.getTimestamp() * 1000L)
//        );
//
//        // 分组聚合开窗
//        DataStream<ItemViewCount> windowAggregateStream = watermarkDataStream.filter(data -> "pv".equals(data.getBehavior()))
//                .keyBy(UserBehavior::getItemId)
//                .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5))) // 滑动窗口实现以每小时分桶、每五分钟滑动
//                // aggregate方法Prefix部分 —— 聚合函数
//                // aggregate方法Suffix部分 —— 窗口函数
//                .aggregate(new HotItemOperators.UserBehaviorAggregateFunctionImpl(),
//                        new HotItemOperators.ItemViewCountWindowFunctionImpl());
//
//
//        SingleOutputStreamOperator<String> resultDataStream = windowAggregateStream.keyBy(ItemViewCount::getTimstamp)
//                .process(new HotItemOperators.ItemViewCountKeyedProcessFunctionImpl(topNumber));
//
//        resultDataStream.print();
//
//        streamEnv.execute();
//    }
}
